package com.rr.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

/**
 * @author zhuangcanrun
 * @Date 2022/10/07/15:15
 * @description
 **/

@Configuration
public class PrimaryMqConfig {
    @Bean(name = "connectionFactory")
    @Primary
    public ConnectionFactory firstConnectionFactory(@Value("${spring.rabbitmq.host}") String host,
                                                    @Value("${spring.rabbitmq.port}") int port,
                                                    @Value("${spring.rabbitmq.virtual-host}") String virtualHost,
                                                    @Value("${spring.rabbitmq.username}") String username,
                                                    @Value("${spring.rabbitmq.password}") String password) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;
    }

    @Primary
    @Bean
    public RabbitTemplate rabbitTemplate(@Qualifier("connectionFactory") ConnectionFactory connectionFactory,
                                         MessageConverter messageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter);
        return rabbitTemplate;
    }

    @Primary
    @Bean
    public RabbitAdmin rabbitAdmin(@Qualifier("connectionFactory") ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter(new ObjectMapper());
    }

    @Bean
    public CustomRabbitListenerContainerFactory rabbitListenerContainerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            ConnectionFactory connectionFactory,
            MessageConverter messageConverter) {
        MessagePostProcessor messagePostProcessor = message -> {
            // 2021-01-03 不再记录接受参数，与skywalking-rabbitmq有冲突，不能包含在skywalking的traceId里面
            message.getMessageProperties().setContentType("application/json");
            return message;
        };
        CustomRabbitListenerContainerFactory factory = new CustomRabbitListenerContainerFactory(messagePostProcessor);
        factory.setMessageConverter(messageConverter);
        configurer.configure(factory, connectionFactory);
        return factory;
    }

    @Bean(name = "customSimpleConsumerFactory")
    public SimpleRabbitListenerContainerFactory customSimpleConsumerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, @Qualifier("connectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory container = new SimpleRabbitListenerContainerFactory();
        container.setRecoveryInterval(10000L);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        configurer.configure(container, connectionFactory);
        return container;
    }

    public static class CustomRabbitListenerContainerFactory
            extends org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory {

        private final MessagePostProcessor[] messagePostProcessor;

        public CustomRabbitListenerContainerFactory(MessagePostProcessor... messagePostProcessor) {
            super();
            this.messagePostProcessor = messagePostProcessor;
        }

        @Override
        protected void initializeContainer(SimpleMessageListenerContainer instance, RabbitListenerEndpoint endpoint) {
            super.initializeContainer(instance, endpoint);
            instance.addAfterReceivePostProcessors(messagePostProcessor);
        }

    }
}
